package com.belf.console.agent.migrate;

import belf.migrate.engine.admin.taskconf.SinkConf;
import belf.migrate.engine.admin.taskconf.SourceConf;
import belf.migrate.engine.admin.taskconf.TaskConf;
import belf.migrate.engine.admin.taskconf.typemapping.DataTypeMapping;
import belf.migrate.engine.admin.taskconf.typemapping.FieldTypeMapping;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.belf.console.domain.*;
import com.belf.console.domain.entity.ColumnSyncConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.belf.common.constant.BelfCodeConstants.BELF_DATA_CONF_MODULE_LINE_ON;

/**
 * BekfTaskConfBuild
 *
 * @author l
 * @since 2024/6/27 13:12
 */

public class BuildTaskConf {

    private static final Logger logger = LoggerFactory.getLogger(BuildTaskConf.class.getName());
    public static TaskConf buildTaskConf(DataCheckTask taskData) {
        logger.info("开始构建任务配置");
        TaskConf taskConf = new TaskConf();
        // 构建taskConf 基本信息
        taskConf.setId(taskData.getId());
        taskConf.setCreateDate(taskData.getCreateTime().getTime());
//        taskConf.setOwner(taskData.getCreateBy());
        taskConf.setTaskName(taskData.getTaskName());
        taskConf.setTaskType(taskData.getTaskType());
        taskConf.setTaskDesc(taskData.getDsc());

        // 构建source
        List<DataSourceInfo> dataSourceInfoList = taskData.getDataSourceInfoList();

        if (dataSourceInfoList != null && dataSourceInfoList.size() > 0) {
            for (DataSourceInfo dataSourceInfo : dataSourceInfoList) {
                if (dataSourceInfo.getId().equals(taskData.getSourceId())){
                    SourceConf sourceConf = buildSourceConf(dataSourceInfo);
                    taskConf.setSourceConf(sourceConf);
                }
                if (dataSourceInfo.getId().equals(taskData.getSinkId())) {
                    //  连接器
                    SinkConf sinkConf = buildSinkConf(dataSourceInfo);
                    taskConf.setSinkConf(sinkConf);
                }
            }
        }
        return taskConf;
    }

    public static TaskConf buildTaskConf(DataTaskConf taskData) {
        logger.info("开始构建任务配置");
        TaskConf taskConf = new TaskConf();
        // 构建taskConf 基本信息
        taskConf.setId(taskData.getId());
        taskConf.setCreateDate(taskData.getCreateTime().getTime());
//        taskConf.setOwner(taskData.getCreateBy());
        taskConf.setTaskName(taskData.getTaskName());
        taskConf.setTaskType(taskData.getTaskType());
        taskConf.setTaskDesc(taskData.getDsc());

        // 构建source
        List<DataSourceInfo> dataSourceInfoList = taskData.getDataSourceInfoList();

        if (dataSourceInfoList != null && dataSourceInfoList.size() > 0) {
            for (DataSourceInfo dataSourceInfo : dataSourceInfoList) {
                if (dataSourceInfo.getId().equals(taskData.getSourceId())){
                    SourceConf sourceConf = buildSourceConf(dataSourceInfo);
//                    sourceConf.getConnectionConf().setOwner(taskData.getSourceOwner());
                    taskConf.setSourceConf(sourceConf);
                }
                if (dataSourceInfo.getId().equals(taskData.getSinkId())) {
                    //  连接器
                    SinkConf sinkConf = buildSinkConf(dataSourceInfo);
                    sinkConf.setReaderBatch(Math.toIntExact(taskData.getReaderBatch()));
//                    sinkConf.getConnectionConf().setOwner(taskData.getSinkOwner());
                    taskConf.setSinkConf(sinkConf);
                }
            }
        }
        // 同步配置
        if (taskData.getSyncStatus() != null && taskData.getSyncStatus() == BELF_DATA_CONF_MODULE_LINE_ON){
            taskConf.setCdc(conversionSyncConf(taskData.getSyncConf()));
        }
        if (taskData.getBatchStatus() != null && taskData.getBatchStatus() == BELF_DATA_CONF_MODULE_LINE_ON){
            List<DataBatchConf> jdbc = taskData.getJdbc();
            List<DataBatchConf> tableBatchConf = taskData.getTableBatchConf();
            jdbc.addAll(tableBatchConf);
            for (DataBatchConf dataBatchConf : jdbc){
                String mapping = taskData.getMapping();
                JSONObject jsonObject = JSONObject.parseObject(mapping);
                List<ColumnSyncConf> list = jsonObject.getList(dataBatchConf.getTableName(), ColumnSyncConf.class);
                dataBatchConf.setColumns(list);
            }


            taskConf.setJdbc(conversionJdbcConf(jdbc));
        }
        // 字段映射配置
        if ( (taskData.getSyncStatus() != null && taskData.getSyncStatus() == BELF_DATA_CONF_MODULE_LINE_ON) ||
                (taskData.getBatchStatus() != null && taskData.getBatchStatus() == BELF_DATA_CONF_MODULE_LINE_ON)){
            taskConf.setMapping(conversionMappingConf(taskData.getMapping()));
        }
        // 字段类型映射配置
        if (taskData.getTypeStatus() != null && taskData.getTypeStatus() == BELF_DATA_CONF_MODULE_LINE_ON) {
            taskConf.setDataTypeMappings(conversionDBTypeMappingConf(taskData.getDbTypeMapping()));
            taskConf.setTableTypeMappings(conversionTableTypeMappingConf(taskData.getTableTypeMapping()));
        }
//        taskConf.setSourceConf(sourceConf);
        return taskConf;
    }

    private static Map<String, List<ColumnSyncConf>> conversionMappingConf(String columnSyncConfMa) {
        return JSONObject.parseObject(columnSyncConfMa, Map.class);
    }

    private static List<DataTypeMapping> conversionDBTypeMappingConf(String typeMappingConf) {
        List<DataTypeMapping> dataTypeMappings = JSONArray.parseArray(typeMappingConf, DataTypeMapping.class);
        return dataTypeMappings;
    }

    private static Map<String, List<FieldTypeMapping>> conversionTableTypeMappingConf(String tableTypeMappingConf) {

        List<FieldTypeMapping> dataTypeMappings = JSONArray.parseArray(tableTypeMappingConf, FieldTypeMapping.class);
        HashMap<String, List<FieldTypeMapping>> tableMapping = new HashMap<>();
        dataTypeMappings.forEach(itm -> {
            if (tableMapping.containsKey(itm.getSource().getTable())) {
                tableMapping.get(itm.getSource().getTable()).add(itm);
            } else {
                List<FieldTypeMapping> fieldTypeMappings = new ArrayList<>();
                fieldTypeMappings.add(itm);
                tableMapping.put(itm.getSource().getTable(), fieldTypeMappings);
            }
        });
        return tableMapping;
    }


    /**
     *  同步任务配置
     * @param taskConf
     */
    public static void BuildTaskConf(List<DataSourceInfo> dataSourceInfoList,TaskConf taskConf,String sourceId,String sinkId ) {

    }

    public static SourceConf buildSourceConf(DataSourceInfo sourceInfo) {
        logger.info("开始构建数据源配置");
        SourceConf conf = new SourceConf();
        //  连接器
        conf.setCatalogName(sourceInfo.getType());
        conf.setSchemaName(sourceInfo.getSchemaName());
        conf.setDatabaseName(sourceInfo.getDatabaseName());
        conf.setFieldCase(sourceInfo.getFieldCase());
        //  区分大小写
        if (sourceInfo.getCaseSensitive() != null) {
//            conf.setCaseSensitive(sourceInfo.getCaseSensitive().equals(BELF_DATA_CASE_SENSITIVE_TRUE));
            conf.setCaseSensitive(Boolean.parseBoolean(sourceInfo.getCaseSensitive()));
        }
        if (sourceInfo.getQuote() != null) {
//            conf.setQuote(sourceInfo.getQuote().equals(BELF_DATA_CASE_SENSITIVE_TRUE));
            conf.setQuote(Boolean.parseBoolean(sourceInfo.getQuote()));
        }
        conf.setConnectionConf(BuildConnect.connectionConf(sourceInfo));
        return conf;
    }

    public static SinkConf buildSinkConf(DataSourceInfo sourceInfo) {
        logger.info("开始构建数据源配置");
        SinkConf conf = new SinkConf();
        conf.setCatalogName(sourceInfo.getType());
        conf.setSchemaName(sourceInfo.getSchemaName());
        conf.setDatabaseName(sourceInfo.getDatabaseName());
        conf.setFieldCase(sourceInfo.getFieldCase());
        //   区分大小写
        if (sourceInfo.getCaseSensitive() != null) {
//            conf.setCaseSensitive(sourceInfo.getCaseSensitive().equals(BELF_DATA_CASE_SENSITIVE_TRUE));
            conf.setCaseSensitive(Boolean.parseBoolean(sourceInfo.getCaseSensitive()));
        }
        if (sourceInfo.getQuote() != null) {
//            conf.setQuote(sourceInfo.getQuote().equals(BELF_DATA_CASE_SENSITIVE_TRUE));
            conf.setQuote(Boolean.parseBoolean(sourceInfo.getQuote()));
        }
        conf.setConnectionConf(BuildConnect.connectionConf(sourceInfo));
        return conf;
    }


    public static HashMap<String, String> conversionSyncConf( String syncConf ) {
        HashMap<String, String> cdcMap = new HashMap<>();
        if (syncConf != null) {
            JSONObject dataCdcConf = JSONObject.parseObject(syncConf);
            //fastjson解析方法
            for (Map.Entry<String, Object> entry : dataCdcConf.entrySet()) {
                System.out.println("key值="+entry.getKey());
                System.out.println("对应key值的value="+entry.getValue());
                cdcMap.put(entry.getKey(),entry.getValue().toString());
            }
            return cdcMap;
/*                cdcMap.put("connector.class",dataCdcConf.getConnectorClass());
                cdcMap.put("offset.flush.interval.ms",dataCdcConf.getFlushInterval());
                cdcMap.put("topic.prefix",dataCdcConf.getTopic());
                cdcMap.put("database.server.id",dataCdcConf.getServerId());
                cdcMap.put("name",dataCdcConf.getServerName());
                cdcMap.put("table.include.list",dataCdcConf.getTableList());
                cdcMap.put("database.include.list",dataCdcConf.getDatabaseList());
                cdcMap.put("snapshot.mode",dataCdcConf.getSnapshotMode());*/
        }
        return null;
    }

    public static  List<TableSyncConf> conversionJdbcConf(List<DataBatchConf> jdbcConf ) {
        List<TableSyncConf>  jdbc= new ArrayList<>();
        if (jdbcConf != null) {
            jdbcConf.forEach(itm -> {
                TableSyncConf conf = new TableSyncConf();
                conf.setSelectSQL(itm.getSelectSQL());
                conf.setInsertColumns(itm.getInsertColumns());
                conf.setTableName(itm.getTableName());
                conf.setLast(itm.getLast());
                conf.setPageSize(itm.getPageSize());
                conf.setSlice(itm.getSlice());
                conf.setSliceSize(itm.getSliceSize());
                conf.setSortBy(itm.getSortBy());
                conf.setSyncType(itm.getSyncType());
                jdbc.add(conf);
            });
//            TableSyncConf dataCdcConf = JSONObject.parseObject(jdbcConf,TableSyncConf.class);
//            JSONArray parse = JSONArray.parse(jdbcConf, TableSyncConf.class);
//            List<TableSyncConf> tableSyncConfs = JSONArray.parseArray(jdbcConf, TableSyncConf.class);

/*                cdcMap.put("connector.class",dataCdcConf.getConnectorClass());
                cdcMap.put("offset.flush.interval.ms",dataCdcConf.getFlushInterval());
                cdcMap.put("topic.prefix",dataCdcConf.getTopic());
                cdcMap.put("database.server.id",dataCdcConf.getServerId());
                cdcMap.put("name",dataCdcConf.getServerName());
                cdcMap.put("table.include.list",dataCdcConf.getTableList());
                cdcMap.put("database.include.list",dataCdcConf.getDatabaseList());
                cdcMap.put("snapshot.mode",dataCdcConf.getSnapshotMode());*/
            return jdbc;
        }
        return null;
    }
}
